package com.cyf.mq.producer;

import com.cyf.mq.constant.MqConstant;
import com.cyf.mq.pojo.OrderInfo;
import com.cyf.mq.util.JacksonUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.MessageQueueSelector;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageQueue;

import java.util.Arrays;
import java.util.List;

/**
 * @author chenyifan
 * @create 2025-02-18 16:23
 */
@Slf4j
public class OrderProducer {

    public static void main(String[] args) throws Exception {
        DefaultMQProducer producer = new DefaultMQProducer(MqConstant.MESSAGE_PRODUCER_GROUP_A);
        producer.setNamesrvAddr(MqConstant.NAME_SERVER_ADDR);
        producer.start();

        List<OrderInfo> orderInfos = Arrays.asList(
                new OrderInfo("1111", "下单"),
                new OrderInfo("1111", "短信"),
                new OrderInfo("1111", "物流"),
                new OrderInfo("2245", "下单"),
                new OrderInfo("2245", "短信"),
                new OrderInfo("2245", "物流")
        );

        // 发送顺序消息，发送时要确保有序，并且要发送到同一个队列下面去
        for (OrderInfo orderInfo : orderInfos) {
            Message message = new Message(MqConstant.MESSAGE_TOPIC_King, JacksonUtil.toJsonString(orderInfo).getBytes());
            // 相同的订单号去相同的队列
            SendResult sendResult = producer.send(message, new MessageQueueSelector() {
                @Override
                public MessageQueue select(List<MessageQueue> list, Message message, Object arg) {
                    // arg 就是 send 方法传入的第三个参数
                    int index = arg.hashCode() % list.size();
                    return list.get(index);
                }
            }, orderInfo.getOrderId());
            log.info("{} -> 发送成功：{}", orderInfo, sendResult.getMessageQueue().getQueueId());
        }
        producer.shutdown();
    }
}